package com.hsc.www.rawTest.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;


public class RPCClient {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private Channel channel;
    private String replyQueueName;
    private final Map<String, BlockingQueue<byte[]>> completionQueueMap = new ConcurrentHashMap<>();

    public RPCClient() throws Exception {

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setVirtualHost("my_vhost");
        connectionFactory.setPort(5672);
        // 创建一个连接
        Connection connection = connectionFactory.newConnection();
        // 创建一个频道
        channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(RPC_QUEUE_NAME, true, false, false, null);
        //为每一个客户端获取一个随机的回调队列
        replyQueueName = channel.queueDeclare().getQueue();
        //消费者与队列关联
        channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                BlockingQueue blockingQueue = completionQueueMap.get(properties.getCorrelationId());
                if (blockingQueue != null) {
                    blockingQueue.add(body);
                }
            }
        });
    }


    public String add(int arg1, int arg2) {

        String correlationId = UUID.randomUUID().toString();
        BlockingQueue<byte[]> blockingQueue = new LinkedBlockingQueue<>(1);
        completionQueueMap.put(correlationId, blockingQueue);

        try {
            //设置replyTo和correlationId属性值
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .correlationId(correlationId)
                    .replyTo(replyQueueName)
                    .build();
            String message = arg1 + "," + arg2;
            //发送消息到rpc_queue队列
            channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes("UTF-8"));

            byte[] body = blockingQueue.take();
            return new String(body, "UTF-8");
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            completionQueueMap.remove(correlationId);
        }


    }

    public static void main(String[] args) throws Exception {

       ExecutorService threadPool= Executors.newFixedThreadPool(100);
        RPCClient client = new RPCClient();
        for (int i = 0; i < 100; i++) {
            for (int j = 0; j < 100; j++) {
                int ii = i, jj = j;
                threadPool.submit(() -> {
                    System.out.println(ii + " + " + jj + " = " + client.add(ii, jj));
                });
            }
        }

    }
}
